#ifndef __MQ_CONSUMER_
#define __MQ_CONSUMER_

#include "../mqcommon/mq_log.hpp"
#include "../mqcommon/mq_tools.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include <string>
#include <mutex>
#include <unordered_map>
#include <memory>
#include <functional>

namespace MQ
{

    using ConsumerCallback = std::function<void(const std::string &, const MQ::BasicProperties *, const std::string &)>;

    struct Consumer
    {
        using ptr = std::shared_ptr<Consumer>;

        std::string _tag;
        std::string _qname;
        bool _auto_ack;
        ConsumerCallback _cb;

        Consumer()
        {
            LOG(DEBUG, "new Consumer %p\n", this);
        }

        Consumer(const std::string &tag, const std::string &qname, bool auto_ack, ConsumerCallback cb)
            : _tag(tag), _qname(qname), _auto_ack(auto_ack), _cb(cb)
        {
            LOG(DEBUG, "new Consumer %p\n", this);
        }

        ~Consumer()
        {
            LOG(DEBUG, "delete Consumer %p\n", this);
        }
    };

    class QueueConsumer
    {
    public:
        using ptr = std::shared_ptr<QueueConsumer>;

        QueueConsumer()
        {
        }

        QueueConsumer(const std::string &qname)
            : _qname(qname), _rr_seq(0)
        {
        }

        // 新增
        Consumer::ptr create(const std::string &tag, const std::string &qname, bool ack_flag, const ConsumerCallback &cb)
        {
            std::unique_lock<std::mutex> lock(_mutex);

            for (auto &con : _consumers)
            {
                if (con->_tag == tag)
                {
                    LOG(WARNING, "create 已经存在该消费者!!!\n");
                    return Consumer::ptr();
                }
            }

            Consumer::ptr cp = std::make_shared<Consumer>(tag, qname, ack_flag, cb);
            _consumers.push_back(cp);
            return cp;
        }

        // 删除
        void remove(const std::string &tag)
        {
            std::unique_lock<std::mutex> lock(_mutex);

            for (auto it = _consumers.begin(); it != _consumers.end(); ++it)
            {
                if ((*it)->_tag == tag)
                {
                    _consumers.erase(it);
                    return;
                }
            }
        }

        // 轮转
        Consumer::ptr RR()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (_consumers.size() == 0)
            {
                LOG(WARNING, "消费者数量为0!!!\n");
                return Consumer::ptr();
            }
            int seq = _rr_seq % _consumers.size();
            ++_rr_seq;
            return _consumers[seq];
        }

        // 是否为空
        bool empty()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _consumers.size() == 0;
        }

        // 是否存在
        bool exists(const std::string &tag)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto it = _consumers.begin(); it != _consumers.end(); ++it)
            {
                if ((*it)->_tag == tag)
                {
                    return true;
                }
            }
            return false;
        }

        // 清理
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _consumers.clear();
            _rr_seq = 0;
        }

    private:
        std::string _qname;
        std::mutex _mutex;
        std::vector<Consumer::ptr> _consumers;
        uint64_t _rr_seq;
    };

    class ConsumerManager
    {
    public:
        using ptr = std::shared_ptr<ConsumerManager>;

        ConsumerManager()
        {
        }

        void initQueueConsumer(const std::string &qname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _qconsumers.find(qname);
            if (it != _qconsumers.end())
            {
                LOG(WARNING, "init已经存在该queueConsumer!!!\n");
                return;
            }
            QueueConsumer::ptr qp = std::make_shared<QueueConsumer>(qname);
            _qconsumers.insert(std::make_pair(qname, qp));
        }

        void destoryQueueConsumer(const std::string &qname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _qconsumers.find(qname);
            if (it == _qconsumers.end())
            {
                LOG(WARNING, "destory不存在该queueConsumer!!!\n");
                return;
            }
            _qconsumers.erase(qname);
        }

        Consumer::ptr create(const std::string &tag, const std::string &qname, bool ack_flag, const ConsumerCallback &cb)
        {
            // 获取队列的消费者管理单元句柄，通过句柄完成新建
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(qname);
                if (it == _qconsumers.end())
                {
                    LOG(WARNING, "没有找到队列 %s 的消费者管理句柄！", qname);
                    return Consumer::ptr();
                }
                qcp = it->second;
            }
            return qcp->create(tag, qname, ack_flag, cb);
        }

        void remove(const std::string &tag, const std::string &qname)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(qname);
                if (it == _qconsumers.end())
                {
                    LOG(WARNING, "没有找到队列 %s 的消费者管理句柄！", qname);
                    return;
                }
                qcp = it->second;
            }
            return qcp->remove(tag);
        }

        Consumer::ptr RR(const std::string &qname)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(qname);
                if (it == _qconsumers.end())
                {
                    LOG(WARNING, "没有找到队列 %s 的消费者管理句柄！", qname);
                    return Consumer::ptr();
                }
                qcp = it->second;
            }
            return qcp->RR();
        }

        bool empty(const std::string &qname)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(qname);
                if (it == _qconsumers.end())
                {
                    LOG(WARNING, "没有找到队列 %s 的消费者管理句柄！", qname);
                    return false;
                }
                qcp = it->second;
            }
            return qcp->empty();
        }

        bool exists(const std::string &tag, const std::string &qname)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(qname);
                if (it == _qconsumers.end())
                {
                    LOG(WARNING, "没有找到队列 %s 的消费者管理句柄！", qname);
                    return false;
                }
                qcp = it->second;
            }
            return qcp->exists(tag);
        }

        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _qconsumers.clear();
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, QueueConsumer::ptr> _qconsumers;
    };
}
#endif